package com.oops.service.impl;

import com.oops.model.Position;
import com.oops.model.PositionPageDTO;
import com.oops.service.IPositionService;
import com.oops.support.MySqlOperator;
import com.oops.support.PositionMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.sql.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

import static com.oops.support.MySqlOperator.conn;

@Slf4j
@Service
public class IPositionServiceImpl implements IPositionService {
    private RestHighLevelClient restHighLevelClient;

    public IPositionServiceImpl(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }

    @Override
    public void transferFromMySql2Es() {
        BulkProcessor bulkProcessor = getBulkProcessor();
        final Connection connection = MySqlOperator.getConnection();
        ResultSet resultSet = null;
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement("select * from lg.position ",
                    ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY);
            preparedStatement.setFetchSize(20);
            resultSet = preparedStatement.executeQuery();
            final ResultSetMetaData metaData = resultSet.getMetaData();

            int count = 0;
            String c = null;
            String v = null;
            List<Map<String, String>> esDataList = new ArrayList<>();
            Map<String, String> map = new HashMap<>();
            while (resultSet.next()) {
                map = new HashMap<String, String>(128);
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    c = metaData.getColumnName(i);
                    v = resultSet.getString(c);
                    map.put(c, v);
                }
                log.debug("map:{}", map);
                esDataList.add(map);


                // 10000条数据写入一次es
                if (count % 10000 == 0) {
                    // 将数据添加到bulkprocessor中
                    for (Map<String, String> stringStringMap : esDataList) {
                        bulkProcessor.add(new IndexRequest("test_position").source(stringStringMap));
                    }
                    esDataList.clear();
                    map.clear();
                }
            }
            // 将数据添加到bulkprocessor中
            for (Map<String, String> stringStringMap : esDataList) {
                bulkProcessor.add(new IndexRequest("test_position").source(stringStringMap));
            }
            esDataList.clear();
            map.clear();

            //将数据刷新到es。不一定会立即生效，取决于bulkProcessor的刷新时间设置。
            bulkProcessor.flush();

        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != resultSet) {
                    resultSet.close();
                }
                if (null != preparedStatement) {
                    preparedStatement.close();
                }
                conn.close();
                boolean terminatedFlag = bulkProcessor.awaitClose(150L, TimeUnit.SECONDS);
                log.info("terminatedFlag:{}", terminatedFlag);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }


    }

    @Override
    public List<Position> queryByConditions(PositionPageDTO param) throws IOException {
        log.info("参数 param:{}", param);
        final List<Position> objects = new ArrayList<>();
        final SearchRequest searchRequest = new SearchRequest("test_position");
        final SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(param.getSize());
        searchSourceBuilder.from((param.getPage() - 1) * param.getSize());


        final MatchQueryBuilder matchQueryBuilder = QueryBuilders.matchQuery(param.getPositionName() == null ? "positionAdvantage" : "positionName",
                param.getPositionName() == null ? param.getPositionAdvantage() : param.getPositionName());
        searchSourceBuilder.query(matchQueryBuilder);
        searchSourceBuilder.timeout(new TimeValue(5, TimeUnit.SECONDS));

        searchRequest.source(searchSourceBuilder);
        final SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        final SearchHit[] hits = searchResponse.getHits().getHits();
        final TotalHits totalHits = searchResponse.getHits().getTotalHits();
        final long value = totalHits.value;
        log.info("结果数量 value:{}", value);


        for (SearchHit hit : hits) {
            final Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            final Position position = PositionMapper.INSTANCE.map2Position(sourceAsMap);
            objects.add(position);
        }
        return objects;
    }


    public BulkProcessor getBulkProcessor() {
        BulkProcessor bulkProcessor = null;

        final BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long l, BulkRequest bulkRequest) {
                log.debug("beforeBulk bulkRequest.numberOfActions:{}", bulkRequest.numberOfActions());
            }

            @Override
            public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
                log.debug("afterBulk success bulkRequest.numberOfActions:{}", bulkRequest.numberOfActions());

            }

            @Override
            public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
                log.error("afterBulk failed ", throwable);

            }
        };
        BiConsumer<BulkRequest, ActionListener<BulkResponse>> consumer =
                (bulkRequest, bulkListener) -> restHighLevelClient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkListener);
        final BulkProcessor.Builder builder = BulkProcessor.builder(consumer, listener);
        //根据当前添加的操作数设置何时刷新新的批量请求。默认值为1000。可以设置为-1以禁用它
        builder.setBulkActions(5000);
        //根据当前添加的操作的大小设置何时刷新新的批量请求。默认为5mb。可以设置为-1以禁用它。
        builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
        // 并发请求数
        builder.setConcurrentRequests(10);
        //设置刷新间隔，如果间隔已过，则刷新任何挂起的批量操作。默认为未设置。注意，setBulkActions（int）和setBulkSize（ByteSizeValue）都可以设置为-1，刷新间隔设置允许对批量操作进行完全异步处理。
        builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));
        bulkProcessor = builder.build();

        return bulkProcessor;
    }
}
